#include "libcomm.h"

bool TaskResult::Join(CommType* result, unsigned int timeoutMs)
{
	if(m_sem.WaitSignal(timeoutMs))
	{
		if(NULL != result)
			*result = m_result;
		return true;
	}
	return false;
}

TaskItem::TaskItem():m_feedback(new TaskResult()){}


//////////////////////////////////////////////////////////////////////////

Task::Task(int threadsCount,const char* taskName)
	: ThreadPool(threadsCount, taskName?taskName:"task"),m_itemList(new CustomListReleaser<TaskItem*>()),m_running(0)
{
	this->Start();
}

Task::~Task(void)
{
	this->Stop(true);
	m_itemList.Clear();
}

void Task::Start()
{
	if(!m_running)
	{
		m_running = 1;
		ThreadPool::Start();
	}
	
}

void Task::Stop(bool cancelThread)
{
	if(m_running)
	{
		m_running = 0;
		ThreadPool::Stop(cancelThread);
	}	
}

SharedPtr<TaskResult> Task::Schedule(TaskItem* item)
{
	SharedPtr<TaskResult> feedback = item->m_feedback;
	m_itemList.Append(item);
	m_sem.SetSignal();
	return feedback;
}

#ifdef __GNUC__
static void __wait_cleanup(void *arg)
{    
	TaskItem* item = *((TaskItem**)arg);
	if(NULL != item)
	{
		LOGWARN("thread(%d) was canceled, delete TaskItem instance to avoid memory leak!",Thread::Id());
		item->Release();
	}
}
#endif

int Task::Process(int index)
{
	index = 0;
	while (m_running)
	{
		TaskItem* item = NULL;
#ifdef __GNUC__
		pthread_cleanup_push(__wait_cleanup, &item);
#endif
		if(m_sem.WaitSignal(1000))
		{
			m_itemList.Pop(item);
			MYLIB_ASSERT_NOTNULL(item);
			item->m_feedback->m_result = item->Do();
			item->m_feedback->m_sem.SetSignal();
			item->Release();
			item = NULL;
		}
#ifdef __GNUC__
		pthread_cleanup_pop(0);
#endif
	}
	return 0;
}



class DefaultTaskItem : public TaskItem
{
public:
	Task::WorkProc m_proc;
	CommType m_arg;
	DefaultTaskItem(Task::WorkProc proc, const CommType& arg)
		:m_proc(proc),m_arg(arg){}
	virtual CommType Do()
	{
		return m_proc(m_arg);
	}
};


SharedPtr<TaskResult> Task::Schedule(Task::WorkProc proc, const CommType& arg)
{
	MYLIB_ASSERT(NULL != proc);
	return Schedule(new DefaultTaskItem(proc,arg));
}